package operator.condition;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import util.RxJavaHelper;
import util.ThreadHelper;

class AmbOperator {

    public static void main(String[] args) {
//        amb();
//        ambArray();
        ambWith();
    }

    /**
     * 只接收第一个发送事件的Observable，包括onNext、onError、onComplete
     * 接收一个集合
     */
    static void amb() {
        List<Observable<Integer>> observables = new ArrayList<>();
        Observable<Integer> o1 = Observable.just(1).delay(100, TimeUnit.SECONDS);
        Observable<Integer> o2 = Observable.just(1, 2, 3);
        observables.add(o1);
        observables.add(o2);
        Observable.amb(observables)
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }

    /**
     * 只接收第一个发送事件的Observable，包括onNext、onError、onComplete
     * 接收多个Observable
     */
    static void ambArray() {
        Observable<Integer> o1 = Observable.just(4).delay(100, TimeUnit.SECONDS);
        Observable<Integer> o2 = Observable.just(4, 5, 6);
        Observable.ambArray(o1, o2)
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }

    static void ambWith() {
        Observable<Integer> o1 = Observable.just(7).delay(100, TimeUnit.SECONDS);
        Observable<Integer> o2 = Observable.just(7, 8, 9);
        o1.ambWith(o2)
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }


}
